package com.foreveross.taskservice.taskcache.redis.core;

import java.util.Set;
import java.util.UUID;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;

import com.foreveross.taskservice.taskcache.redis.core.exception.WaitLockTimeoutException;
import com.foreveross.taskservice.taskcache.redis.util.JedisUtil;
import com.foreveross.taskservice.taskcache.redis.util.RedisSerializeUtil;
import com.google.common.collect.Sets;
/**
 * redis模版
 * @author Administrator
 *
 */
public class TaskRedisTemplate {
	final static long Millisecond = 1000L;
	final static String NX = "NX";
	final static String XX = "XX";
	final static String EX = "EX";
	final static String PX = "PX";
	
	final static byte[] NX_BYTE=RedisSerializeUtil.serializeString(NX);
	final static byte[] XX_BYTE=RedisSerializeUtil.serializeString(XX);
	final static byte[] EX_BYTE=RedisSerializeUtil.serializeString(EX);
	final static byte[] PX_BYTE=RedisSerializeUtil.serializeString(PX);
	
	final static long ONE_MILLI_NANOS=1000000L;//一毫秒的毫微秒
	private final static long THREAD_SLEEP=1000L;//1s轮询一次锁
	
	
	private Logger log=LoggerFactory.getLogger(TaskRedisTemplate.class);

	/**
	 * 回调执行，自动打开释放连接
	 * @param action
	 * @return
	 * @throws Exception
	 */
	public <T> T execute(TaskRedisCallback<T> action)throws Exception{
		Jedis jedis=null;
		boolean good=false;
		T result;
		try{
			jedis=JedisUtil.getResource();
			result=action.doInRedis(jedis);
			good=true;//如果顺利执行，表示jedis不存在通信问题
		}finally{
			returnResource(!good, jedis);
		}
		return result;
	}
	
	
	/**
	 * 支持跨锁的回调
	 * 当等锁超时的时候，会抛出异常
	 * 当执行过程过长超时的时候,如果有其他竞争资源，会导致事务提交失败为null
	 * @param action
	 * @param lockKey
	 * @param waitTimeoutSeccond
	 * @param lockTimeoutSeccond
	 * @return
	 * @throws Exception
	 */
	public <T> T executeLock(TaskRedisCallback<T> action,
			String lockKey,long waitLockExpire,long lockKeyExpire)throws Exception{
		Set<String> keys=Sets.newHashSet(lockKey);
		return executeLocks(action,keys,waitLockExpire,lockKeyExpire);
	}
	
	/**
	 * 支持跨锁的回调，而且支持多锁
	 * @param action
	 * @param lockKeys
	 * @param waitLockExpire
	 * @param lockKeyExpire
	 * @return
	 * @throws Exception
	 */
	public <T> T executeLocks(TaskRedisCallback<T> action,
			Set<String>lockKeys,long waitLockExpire,long lockKeyExpire)throws Exception{
		Jedis jedis=null;
		T result;
		String lockValue=UUID.randomUUID().toString();
		byte[] lockValueByte=RedisSerializeUtil.serializeString(lockValue);
		byte[][] lockKeysByte=buildKeysByte(lockKeys);
		try{
			jedis=JedisUtil.getResource();
			locks(jedis,lockKeysByte , lockValueByte, waitLockExpire, lockKeyExpire, true, true);
			result=action.doInRedis(jedis);
		}finally{
			unlocksAndReturnJedis(jedis,lockKeysByte,lockValueByte);
		}
		return result;
	}
	
	/**
	 * 解锁且正确回收jedis
	 * 目的是当用户调用jedis之后，jedis或者因为管道或者关闭等原因，失效
	 * 所以应该是先尝试解锁，如果是jedis的原因失败，那么应该拿一个新的jedis，解锁且销毁上一个
	 * @param j
	 * @param lockKeys
	 * @param lockValue
	 */
	private void unlocksAndReturnJedis(Jedis j,byte[][] lockKeys,byte[] lockValue){
		boolean good=false;
		try{
			unlocks(j, lockKeys, lockValue);
			j.unwatch();
			good=true;
		}catch(Exception e){
			log.error(String.format("[%s] 解锁失败 %s", j,ExceptionUtils.getFullStackTrace(e)));
			try {
				Thread.sleep(1000L);
			} catch (InterruptedException e1) {}
			unlocksAndReturnJedis(JedisUtil.getResource(),lockKeys,lockValue);
		}finally{
			returnResource(!good,j);
		}
	}
	
	private void returnResource(boolean isDestroy,Jedis j){
		if(!isDestroy){
			JedisUtil.returnResource(j);//回收
		}else{
			JedisUtil.returnBrokenResource(j);//销毁
		}
	}
	
	
	private byte[][] buildKeysByte(Set<String> lockKeys){
		byte[][] result=new byte[lockKeys.size()][];
		int i=0;
		for(String k:lockKeys){
			result[i]=RedisSerializeUtil.serializeString(k);
			i++;
		}
		return result;
	}
	
	
	protected void lock(Jedis j,byte[] lockKey,byte[] lockValue,long waitExpire,long keyExpire,boolean isWatch,boolean expertKeyExpire)throws Exception{
		locks(j, new byte[][]{lockKey}, lockValue, waitExpire, keyExpire, isWatch, expertKeyExpire);
	}
	
	protected void locks(Jedis j,byte[][] lockKeys,byte[] lockValue,long waitExpire,long keyExpire,boolean isWatch,boolean expertKeyExpire)
		throws Exception{
		long start=System.currentTimeMillis();
		for(byte[] lk:lockKeys){
			if(keyExpire<=0){
				throw new WaitLockTimeoutException(RedisSerializeUtil.deserializeString(lk));
			}
			keyExpire=lockAndReturnRemainKeyExpire(j,lk,lockValue,(waitExpire-(System.currentTimeMillis()-start)),keyExpire,isWatch,expertKeyExpire);
			//log.info(String.format("[%s] 加锁:[%s] 估计剩余超时[%s]", j,RedisSerializeUtil.deserializeString(lk),keyExpire));
		}
	}
	
	protected void unlock(Jedis j,byte[] lockKey,byte[] lockValue){
		unlocks(j,new byte[][]{lockKey},lockValue);
	}
	
	protected void unlocks(Jedis j,byte[][] lockKeys,byte[] lockValue){
		String lockValueStr=RedisSerializeUtil.deserializeString(lockValue);
		for(byte[] lk:lockKeys){
			byte[] nowLockValue=j.get(lk);
			if(nowLockValue!=null && RedisSerializeUtil.deserializeString(nowLockValue).equals(lockValueStr)){
				j.del(lk);
				//log.info(String.format("[%s] 解锁:[%s]",j, RedisSerializeUtil.deserializeString(lk)));
			}
		}
	}
	
	
	/**
	 * 取得锁，并且返回当时锁的剩余超时时长
	 * @param j
	 * @param lockKey
	 * @param lockValue
	 * @param waitExpire
	 * @param keyExpire
	 * @param isWatch
	 * @param expertKeyExpire
	 * @return
	 * @throws Exception
	 */
	private long lockAndReturnRemainKeyExpire(
			Jedis j,byte[] lockKey,byte[] lockValue,long waitExpire,long keyExpire,boolean isWatch,boolean expertKeyExpire)throws Exception{
		long nano=System.nanoTime();
		long wn=waitExpire*ONE_MILLI_NANOS;
		boolean lock=false;
		while(System.nanoTime()-nano<wn){
			long expire=expertKeyExpire?(keyExpire-(System.nanoTime()-nano)/ONE_MILLI_NANOS):keyExpire;
			if(j.set(lockKey, lockValue, NX_BYTE, PX_BYTE,expire)!=null){
				if(isWatch){
					j.watch(lockKey);
				}
				lock=true;
				break;
			}
			Thread.sleep(THREAD_SLEEP);
		}
		if(lock){
			//return expertKeyExpire?(keyExpire-(System.nanoTime()-nano)/ONE_MILLI_NANOS):keyExpire;
			return j.pttl(lockKey);
		}else{
			throw new WaitLockTimeoutException(RedisSerializeUtil.deserializeString(lockKey)); 
		}
	}
	
	
}
